/**
 * Alipay.com Inc.
 * Copyright (c) 2004-2020 All Rights Reserved.
 */
package com.dal.flink;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * @author dingaolin
 * @version : DimJoinJob.java, v 0.1 2020年10月10日 下午7:49 dingaolin Exp $
 */
public class DimJoinJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        String sourceSql = "CREATE TABLE datagen (userid int, proctime as PROCTIME())" +
                " WITH (" +
                "'connector' = 'datagen', " +
                "'rows-per-second'='30', " +
                "'fields.userid.kind'='random', " +
                "'fields.userid.min'='1', 'fields.userid.max'='30'" +
                ")";
        tEnv.executeSql(sourceSql);

        String mysqlDDL = "CREATE TABLE dim_mysql (" +
                "  id BIGINT," +
                "  name STRING," +
                "  PRIMARY KEY (id) NOT ENFORCED" +
                ") WITH (" +
                "   'connector' = 'jdbc'," +
                "   'url' = 'jdbc:mysql://localhost:3306/test'," +
                "   'table-name' = 'userinfo'," +
                "   'username' = 'root'," +
                "   'password' = '123456'" +
                ")";
        tEnv.executeSql(mysqlDDL);

        String sql = "SELECT * FROM datagen " +
                "LEFT JOIN dim_mysql FOR SYSTEM_TIME AS OF datagen.proctime " +
                "ON datagen.userid = dim_mysql.id";
        Table table = tEnv.sqlQuery(sql);
        tEnv.toAppendStream(table, Row.class).print();

        env.execute("DimJoinJob");
    }
}